package com.coolpad.job.device.service;

import com.alibaba.fastjson.JSONObject;
import com.coolpad.Utils.ExtAssert;
import com.coolpad.job.device.model.DeviceLocationDTO;
import com.coolpad.job.device.util.Constants;
import com.coolpad.job.device.util.JsonUtils;
import com.coolpad.job.device.util.KafkaMessageUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * 功能描述:<br/>
 *
 * @Author:shengtao_yang
 * @Date: 2020/12/21 13:26
 * @since: 1.0.0
 */
@Slf4j
@Service
public class KafkaSendService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.location.update.topic}")
    public String locationTopic;

    public void sendDeviceLocation(String decryptData) {

        DeviceLocationDTO deviceLocation = JsonUtils.jsonToBean(decryptData, DeviceLocationDTO.class);
        ExtAssert.notNull(deviceLocation,"device location is empty");

        List<JSONObject> messageList = KafkaMessageUtils.convert2Message(deviceLocation);
        messageList.forEach(message -> {
                log.info("[location-collect send] - message:{}",message.toJSONString());
                kafkaTemplate
                    .send(locationTopic, message.getString(Constants.DEVICE_INFO.IMEI1), message.toJSONString())
                    .addCallback(
                            success ->{
                                String topic = success.getRecordMetadata().topic();
                                int partition = success.getRecordMetadata().partition();
                                long offset = success.getRecordMetadata().offset();
                                log.info("[location-collect success] - topic:{}, partition:{}, offset:{}",topic,partition,offset);
                            },
                            failure -> {
                                String failureMessage = failure.getMessage();
                                log.error("[location-collect failure] - exception:{}",failureMessage);
                                throw new RuntimeException("");
                            }
                    );}
        );
    }
}
